import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;

/**
 * Created by taos on 2017/4/1.
 * this program read file from HDFS, implement join merge and put data into Hbase
 */

public class Hw1Grp0 {
    public static void main(String[] args) throws IOException {
        Hw1Grp0 hw1Grp0 = new Hw1Grp0();
        Argments argments = new Argments(args);
        ArrayList<OneResult> tableR, tableS;
//      R=/hw1/lineitem.tbl S=/hw1/orders.tbl join:R0=S0 res:S1,R1,R5

        System.out.println(argments.getFileR());
        System.out.println(argments.getFileS());
        System.out.println(argments.getJoinkeyR());
        System.out.println(argments.getJoinkeyS());
        System.out.println(argments.getValueR());
        System.out.println(argments.getValueS());

        tableR = hw1Grp0.readFile(argments.getFileR(), argments.getJoinkeyR(), argments.getValueR());
        tableS = hw1Grp0.readFile(argments.getFileS(), argments.getJoinkeyS(), argments.getValueS());
        HashMap<String, ArrayList<OneResult>> results = hw1Grp0.hashJoin(tableR, tableS, argments.getJoinkeyR(), argments.getJoinkeyS());
        hw1Grp0.putDataHBase(results, argments.getValueR(), argments.getValueS());
    }

	/**
	* read file and store
     * @param HDFSFilePath file path
     * @param joinkey join key
     * @param valueList join value
     * @return ArrayList every line store one result
	*/
    private ArrayList<OneResult> readFile(String HDFSFilePath, int joinkey, ArrayList<Integer> valueList) throws IOException {
        String file = "hdfs://localhost:9000" + HDFSFilePath;
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(file), conf);
        Path path = new Path(file);
        FSDataInputStream in_stream = fs.open(path);
        BufferedReader in = new BufferedReader(new InputStreamReader(in_stream));
        ArrayList<OneResult> result = new ArrayList<>();
        String s;
        String[] segs;
        OneResult oneResult = null;
        int count = 0;
        while ((s = in.readLine()) != null) {
            segs = s.split("\\|");
            oneResult = new OneResult();
            oneResult.setJoinkey(segs[joinkey]);
            for (int i = 0; i < valueList.size(); i++) {
                oneResult.getValueR().add(segs[valueList.get(i)]);
            }
            result.add(oneResult);
            count++;
        }
        in.close();
        fs.close();
        System.out.printf("%s file : %d", HDFSFilePath, count);
        return result;
    }

	
	/**
	* hash join and store data into map
     * @param tableR table R
     * @param tableS table S
     * @return HashMap store the results
	*/
    private HashMap<String, ArrayList<OneResult>> hashJoin(ArrayList<OneResult> tableR, ArrayList<OneResult> tableS, int joinkeyR, int joinkeyS) {
        HashMap<String, ArrayList<OneResult>> result = new HashMap<>();
        HashMap<String, ArrayList<OneResult>> leftJoinTable = new HashMap<>();

        String joinkey;
        for (int i = 0; i < tableR.size(); i++) {
            joinkey = tableR.get(i).getJoinkey();
            if (leftJoinTable.containsKey(joinkey)) {
                leftJoinTable.get(joinkey).add(tableR.get(i));
            } else {
                leftJoinTable.put(joinkey, new ArrayList<OneResult>());
                leftJoinTable.get(joinkey).add(tableR.get(i));
            }
        }

        for (int i = 0; i < tableS.size(); i++) {
            joinkey = tableS.get(i).getJoinkey();
            if (leftJoinTable.containsKey(joinkey)) {
                for (int j = 0; j < leftJoinTable.get(joinkey).size(); j++) {
                    OneResult oneResult = new OneResult();
                    oneResult.setJoinkey(joinkey);
                    oneResult.setValueR(leftJoinTable.get(joinkey).get(j).getValueR());
                    oneResult.setValueS(tableS.get(i).getValueR());
                    if (result.containsKey(joinkey)) {
                        result.get(joinkey).add(oneResult);
                    } else {
                        result.put(joinkey, new ArrayList<OneResult>());
                        result.get(joinkey).add(oneResult);
                    }
                }
            }
        }
        return result;
    }

	
	/**
	* put data to HBase
     * @param result the merge results
     * @param valueListR table R value
     * @param valueListS table S value
	*/
    private void putDataHBase(HashMap<String, ArrayList<OneResult>> result, ArrayList<Integer> valueListR, ArrayList<Integer> valueListS) throws IOException {
        // if exit delete, than create
        String tableName = "Result";
        HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor cf = new HColumnDescriptor("res");
        htd.addFamily(cf);

        Configuration configuration = HBaseConfiguration.create();
        HBaseAdmin hAdmin = new HBaseAdmin(configuration);
        if (hAdmin.tableExists(tableName)) {
            System.out.println("exit table: " + tableName);
            hAdmin.disableTable(tableName);
            hAdmin.deleteTable(tableName);
        }
        hAdmin.createTable(htd);
        hAdmin.close();

        HTable table = new HTable(configuration, tableName);

        for (String key : result.keySet()) {
            Put put = new Put(key.getBytes());
            if (result.get(key).size() == 1) {
                for (int i = 0; i < valueListR.size(); i++) {
                    put.add("res".getBytes(), ("R" + Integer.toString(valueListR.get(i))).getBytes(), result.get(key).get(0).getValueR().get(i).getBytes());
                }
                for (int i = 0; i < valueListS.size(); i++) {
                    put.add("res".getBytes(), ("S" + Integer.toString(valueListS.get(i))).getBytes(), result.get(key).get(0).getValueR().get(i).getBytes());
                }
            } else {
                for (int i = 0; i < valueListR.size(); i++) {
                    put.add("res".getBytes(), ("R" + Integer.toString(valueListR.get(i))).getBytes(), result.get(key).get(0).getValueR().get(i).getBytes());
                }
                for (int i = 0; i < valueListS.size(); i++) {
                    put.add("res".getBytes(), ("S" + Integer.toString(valueListS.get(i))).getBytes(), result.get(key).get(0).getValueS().get(i).getBytes());
                }

                for (int j = 1; j < result.get(key).size(); j++) {
                    for (int i = 0; i < valueListR.size(); i++) {
                        put.add("res".getBytes(), ("R" + Integer.toString(valueListR.get(i)) + "." + Integer.toString(j)).getBytes(), result.get(key).get(j).getValueR().get(i).getBytes());
                    }
                    for (int i = 0; i < valueListS.size(); i++) {
                        put.add("res".getBytes(), ("S" + Integer.toString(valueListS.get(i)) + "." + Integer.toString(j)).getBytes(), result.get(key).get(j).getValueS().get(i).getBytes());
                    }
                }
            }
            table.put(put);
        }
        table.close();
    }
}

/**
 * this class store the paraments
 */
class Argments {
    private String[] args;
    private String fileR, fileS;
    private int joinkeyR, joinkeyS;
    private ArrayList<Integer> valueR = new ArrayList<>();
    private ArrayList<Integer> valueS = new ArrayList<>();

    public Argments(String[] args) {
        this.args = args;
        this.fileR = args[0].split("=")[1];
        this.fileS = args[1].split("=")[1];
        this.joinkeyR = Integer.parseInt(args[2].split(":")[1].split("=")[0].substring(1));
        this.joinkeyS = Integer.parseInt(args[2].split(":")[1].split("=")[1].substring(1));
        String[] valueList = args[3].split(":")[1].split(",");
        for (String value : valueList) {
            if (value.substring(0, 1).equals("R")) {
                this.valueR.add(Integer.parseInt(value.substring(1)));
            }
            if (value.substring(0, 1).equals("S")) {
                this.valueS.add(Integer.parseInt(value.substring(1)));
            }
        }

    }

    public String getFileR() {
        return fileR;
    }

    public void setFileR(String fileR) {
        this.fileR = fileR;
    }

    public String getFileS() {
        return fileS;
    }

    public void setFileS(String fileS) {
        this.fileS = fileS;
    }

    public int getJoinkeyR() {
        return joinkeyR;
    }

    public void setJoinkeyR(int joinkeyR) {
        this.joinkeyR = joinkeyR;
    }

    public int getJoinkeyS() {
        return joinkeyS;
    }

    public void setJoinkeyS(int joinkeyS) {
        this.joinkeyS = joinkeyS;
    }

    public ArrayList<Integer> getValueR() {
        return valueR;
    }

    public void setValueR(ArrayList<Integer> valueR) {
        this.valueR = valueR;
    }

    public ArrayList<Integer> getValueS() {
        return valueS;
    }

    public void setValueS(ArrayList<Integer> valueS) {
        this.valueS = valueS;
    }
}

/**
 * this class store one row
 */
class OneResult {
    private String joinkey;
    private ArrayList<String> valueR = new ArrayList<>();
    private ArrayList<String> valueS = new ArrayList<>();

    public String getJoinkey() {
        return joinkey;
    }

    public void setJoinkey(String joinkey) {
        this.joinkey = joinkey;
    }

    public ArrayList<String> getValueR() {
        return valueR;
    }

    public void setValueR(ArrayList<String> valueR) {
        this.valueR = valueR;
    }

    public ArrayList<String> getValueS() {
        return valueS;
    }

    public void setValueS(ArrayList<String> valueS) {
        this.valueS = valueS;
    }
}